import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.kou.hadoop.io.MessagePackWritable;
import org.msgpack.type.ValueFactory;


public class MessagePackTestJob extends Configured implements Tool{

	
	private static class MessagePackTestMapper extends Mapper<LongWritable , Text , MessagePackWritable , MessagePackWritable>{
		
		@Override
		public void map(LongWritable key , Text value , Context context)throws IOException , InterruptedException{
			
			MessagePackWritable okey = new MessagePackWritable(ValueFactory.createIntegerValue(key.get()));
			MessagePackWritable ovalue = new MessagePackWritable(ValueFactory.createRawValue(value.toString()));
			context.write(okey , ovalue);
		}
	}
	
	private static class MessagePackTestReducer extends Reducer<MessagePackWritable , MessagePackWritable , LongWritable , Text>{
		
		@Override
		public void reduce(MessagePackWritable key , Iterable<MessagePackWritable> values , Context context)throws IOException , InterruptedException{
			
			Iterator<MessagePackWritable> iterator = values.iterator();
			
			LongWritable okey = new LongWritable(key.get().asIntegerValue().getLong());
			Text ovalue = new Text();
			
			while(iterator.hasNext()){
				ovalue.set(iterator.next().get().asRawValue().getString());
				context.write(okey, ovalue);
			}
		}
	}
    
	@Override
    public int run(String[] args)throws Exception{

    	Job job = new Job();
    	job.setJobName("MessagePackTestJob");
		job.setJarByClass(MessagePackTestJob.class);
	
		job.setMapperClass(MessagePackTestMapper.class);
		job.setReducerClass(MessagePackTestReducer.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job , new Path(args[1]));
		
		job.setMapOutputKeyClass(MessagePackWritable.class);
		job.setMapOutputValueClass(MessagePackWritable.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		
		return (job.waitForCompletion(true)) ? 0 : 1;
    }
    
    public static void main(String[] args)throws Exception{

    	int returnCode = ToolRunner.run(new MessagePackTestJob() , args);
    	System.exit(returnCode);
    }
}